package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo01WordCount {
  def main(args: Array[String]): Unit = {
    // 1、构建Spark环境
    // 配置Spark任务
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo01WordCount") // 设置任务名
    conf.setMaster("local") // 设置Spark的运行方式
    // 创建SparkContext --> Spark程序的入口
    val sc: SparkContext = new SparkContext(conf)

    // 2、构建Spark程序
    // 读取文件
    /**
     * 如果是Local本地模式：该路径为本地的路径，需要在我的电脑里配置系统环境变量HADOOP_HOME
     * 如果是以集群模式运行：该路径为HDFS的路径
     * 不管是哪种方式运行，都是使用Hadoop的那一套读取文件的方式（切片、格式化数据）
     */
    /**
     * RDD：弹性分布式数据集，它不真正存储数据，它只是一个编程模型（统一了所有的方法，所有的Spark的程序都是基于RDD进行处理的）
     * 在使用层面，可以把它当成Scala中的List去使用，List有的常用的方法，RDD都有
     * 例如都有：map、flatMap、filter、foreach、sortBy、groupBy
     * 这些方法在Spark中统称为 算子
     * 算子可以分为两类：转换算子、行为算子
     */
    val lineRDD: RDD[String] = sc.textFile("Spark/data/words.txt")

    //    lineRDD.foreach(println)

    // 将每个单词提取出来
    val wordsRDD: RDD[String] = lineRDD.flatMap(_.split(","))

    // 分区类算子(ByKey类算子) 只能作用在KV格式的RDD上
    //    val kvRDD: RDD[(String, Int)] = wordsRDD.map(word => (word, 1))
    //    kvRDD.groupByKey()

    // 按每个单词进行分组
    val groupRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)


    // 统计每个单词的数量
    val wordCntRDD: RDD[String] = groupRDD.map(kv => s"${kv._1},${kv._2.size}")

    // 将结果输出
    wordCntRDD.foreach(println)

    // 链式调用
    //    sc
    //      .textFile("Spark/data/words.txt")
    //      .flatMap(_.split(","))
    //      .groupBy(word => word)
    //      .map(kv => s"${kv._1},${kv._2.size}")
    //      .foreach(println)

    while (true) {

    }


  }

}
